#include "pool.h"
#include <mqueue.h>
#include <pthread.h>
#include <stdlib.h>

#define MSG_LEN sizeof(Task)

typedef struct TaskStruct
{
    void (*func)(void *);
    void *argv;
} Task;

static mqd_t mq;

static pthread_t *executor_list;
static int executor_count;

void *pool_executor(void *argv)
{
    Task task;
    while (1)
    {
        int result = mq_receive(mq, (char *)&task, MSG_LEN, 0);
        if (result < 0)
        {
            continue;
        }

        task.func(task.argv);
    }
}

int pool_init(int count)
{
    struct mq_attr attr;
    attr.mq_maxmsg = 1000;
    attr.mq_msgsize = MSG_LEN;
    // 初始化消息队列
    mq = mq_open("/task_queue", O_CREAT | O_RDWR, 0644, &attr);
    if (mq < 0)
    {
        return -1;
    }

    executor_count = count;
    executor_list = malloc(sizeof(pthread_t) * executor_count);
    if (!executor_list)
    {
        return -1;
    }

    // 启动后台的executor
    for (size_t i = 0; i < executor_count; i++)
    {
        pthread_create(executor_list + i, NULL, pool_executor, NULL);
    }

    return 0;
}

int pool_registerTask(void (*task_func)(void *), void *argv)
{
    Task task;
    task.func = task_func;
    task.argv = argv;
    mq_send(mq, (char *)&task, MSG_LEN, 0);
    return 0;
}

void pool_deinit()
{
    for (size_t i = 0; i < executor_count; i++)
    {
        pthread_cancel(executor_list[i]);
    }
    for (size_t i = 0; i < executor_count; i++)
    {
        pthread_join(executor_list[i], NULL);
    }
    free(executor_list);
    mq_unlink("/task_queue");
}
